package com.eshipenet.shipnet.equipmentdatahub
import org.apache.flink.api.common.state.{MapStateDescriptor, ValueStateDescriptor}
import org.apache.flink.util.Collector
import play.api.libs.json.JsValue
import org.apache.flink.streaming.api.functions.co.{CoProcessFunction, KeyedBroadcastProcessFunction}

import scala.util.Try

object EdgeChannelDataCoProcessor {
  type E1  = EdgeChannelData
  type E2  = EquipmentWithModel
  type OUT = Try[EquipmentData]
}

import EdgeChannelDataCoProcessor._

class EdgeChannelDataCoProcessor(desc: MapStateDescriptor[String, E2])
    extends KeyedBroadcastProcessFunction[String, E1, E2, OUT] {

  override def processBroadcastElement(value: E2,
                                       ctx: KeyedBroadcastProcessFunction[String, E1, E2, OUT]#Context,
                                       out: Collector[OUT]): Unit = {
    val state = ctx.getBroadcastState(desc)
    state.put(value.id, value)
    out.close()
  }

  override def processElement(value: E1,
                              ctx: KeyedBroadcastProcessFunction[String, E1, E2, OUT]#ReadOnlyContext,
                              out: Collector[OUT]): Unit = {
    val state = ctx.getBroadcastState(desc)
    state.immutableEntries().forEach { kv =>
      val em = kv.getValue
      if (em.isSubscribing(value)) {
        out.collect(em.input(value))
      }
    }
    out.close()
  }

}
